package com.lewin.ringroad.net.handler;

import com.lewin.commons.constants.Time;
import com.lewin.ringroad.dao.CommonMapper;
import com.lewin.ringroad.entity.DigitalTwinPO;
import com.lewin.ringroad.entity.RadarStatisticsDataPO;
import com.lewin.ringroad.net.message.BaseMessage;
import com.lewin.ringroad.net.message.CoilStatisticsMessage;
import com.lewin.ringroad.net.message.QueueLengthMessage;
import com.lewin.ringroad.net.message.RaysunMessage;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * 雷森报文处理器
 *
 * @author Jun
 * @since 1.0.0
 */
@Slf4j
@Component
@ChannelHandler.Sharable
public class RaysunHandler extends SimpleChannelInboundHandler<RaysunMessage> implements Runnable, DisposableBean {

    private final BlockingQueue<BaseMessage> queue = new ArrayBlockingQueue<>(1_000_000);
    public static Map<String, DigitalTwinPO> radarIdAndDTMap;
    public static Map<String, DigitalTwinPO> dtIdAndDTMap;
    private final CommonMapper commonMapper;
    public static final Map<String, RadarStatisticsDataPO> LATEST_HOUR_DATA = new HashMap<>();
    private int latestHour;

    public RaysunHandler(CommonMapper commonMapper) {
        var radarList = commonMapper.allRadar();
        this.commonMapper = commonMapper;
        log.info("radars num: {}", radarList.size());
        radarIdAndDTMap = radarList.stream().collect(Collectors.toMap(DigitalTwinPO::getBizId, Function.identity()));
        dtIdAndDTMap = radarList.stream().collect(Collectors.toMap(t -> String.valueOf(t.getId()), Function.identity()));

        // 获取最近的数据
        var statisticsDataPOS = commonMapper.listByTimes(null, List.of(Time.format(LocalDateTime.now(), "yyyyMMddHH")));
        for (var twinPO : radarList) {
            var bizId = twinPO.getBizId();
            var dataPO = statisticsDataPOS
                    .stream()
                    .filter(t -> bizId.equals(t.getBizId()))
                    .findFirst()
                    .orElseGet(() -> RadarStatisticsDataPO.init(twinPO.getBizId()));
            LATEST_HOUR_DATA.put(bizId, dataPO);
        }

        latestHour = LocalDateTime.now().getHour();

        new Thread(this).start();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RaysunMessage msg) {
        // 打印报文
        var radarId = String.valueOf(msg.getRadarId());
        var twinPO = radarIdAndDTMap.get(radarId);
        if (twinPO == null) {
            log.warn("unknown radar, msg: {}", msg);
            return;
        }
        var baseMessage = msg.toBaseMessage();
        if (baseMessage == null) {
            log.warn("unsupported msg: {}", msg);
            return;
        }
        baseMessage.setBizId(radarId)
                .setDtId(twinPO.getId());
        queue.add(baseMessage);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error(cause.getMessage(), cause);

        ctx.close();
    }

    @Override
    public void run() {
        //noinspection InfiniteLoopStatement
        while (true) {
            try {
                var msg = queue.poll();
                if (msg == null) {
                    msg = queue.take();
                }

                // 检查时间是否变化
                var now = LocalDateTime.now();
                if (now.getHour() != latestHour) {
                    latestHour = now.getHour();
                    // 数据入库
                    var timeAt = Time.format(now, "yyyyMMddHH");
                    LATEST_HOUR_DATA.values().forEach(t -> t.setTimeAt(timeAt));
                    commonMapper.batchInsert(LATEST_HOUR_DATA.values());
                    log.info("数据入库");
                    // 清理数据
                    LATEST_HOUR_DATA.values().forEach(RadarStatisticsDataPO::clear);
                    continue;
                }

                // 只处理线圈统计数据和排队数据
                if (msg instanceof CoilStatisticsMessage csm) {
                    if (!csm.isValidData()) {
                        continue;
                    }
                    String bizId = csm.getBizId();
                    var dataPO = LATEST_HOUR_DATA.get(bizId);
                    if (dataPO == null) {
                        log.warn("radar[{}]不存在", bizId);
                        continue;
                    }
                    var lst = csm.getChannelStatisticsDataList();
                    int traffic = 0;
                    float as = 0;
                    float or = 0;
                    var times = 0;
                    for (var data : lst) {
                        if (!data.isValidData()) {
                            // 确保单通道数据正常
                            continue;
                        }
                        times++;
                        traffic += (data.getATypeTraffic() + data.getBTypeTraffic() + data.getCTypeTraffic());
                        as += data.getAverageSpeed();
                        or += data.getOccupationRate();
                    }
                    if (times != 0) {
                        traffic = traffic / times;
                        as = as / times;
                        or = or / times;
                    }
                    dataPO.setTraffic((traffic + dataPO.getTraffic()) / 2);
                    dataPO.setAverageSpeed((as + dataPO.getAverageSpeed()) / 2);
                    dataPO.setOccupationRate((or + dataPO.getOccupationRate()) / 2);

                    log.info("radarId:{} traffic: {} averageSpeed: {} occupationRate: {} queueLen: {}",
                            bizId, dataPO.getTraffic(), dataPO.getAverageSpeed(), dataPO.getOccupationRate(), dataPO.getQueueLen());
                } else if (msg instanceof QueueLengthMessage qlm) {
                    float averageLen = qlm.averageLen();
                    if (averageLen == 0) {
                        continue;
                    }
                    String bizId = qlm.getBizId();
                    var dataPO = LATEST_HOUR_DATA.get(bizId);
                    if (dataPO == null) {
                        log.warn("radar[{}]不存在", bizId);
                        continue;
                    }
                    dataPO.setQueueLen((dataPO.getQueueLen() + averageLen) / 2);
                    log.info("radar[{}] traffic: {} averageSpeed: {} occupationRate: {} queueLen: {}",
                            bizId, dataPO.getTraffic(),dataPO.getAverageSpeed(), dataPO.getOccupationRate(), dataPO.getQueueLen());
                }
            } catch (Throwable e) {
                log.error("message process failure: {}", e.getMessage(), e);
            }
        }
    }

    @Override
    public void destroy() {
        // 数据入库
        var now = LocalDateTime.now();
        var timeAt = Time.format(now, "yyyyMMddHH");
        LATEST_HOUR_DATA.values().forEach(t -> t.setTimeAt(timeAt));
        commonMapper.batchInsert(LATEST_HOUR_DATA.values());
        log.info("应用关闭, 数据入库完成");
    }
}
